#include <iostream>
#include <vector>
#include <unistd.h>
#include <cstring>
#include <ctime>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <functional>
#include <unordered_map>
#include <sys/epoll.h>
#include <memory>
#include <sys/eventfd.h>
#include <thread>
#include <mutex>
#include <atomic>
#include <sys/timerfd.h>
#include <any>
#include <thread>
#include <condition_variable>
#include <assert.h>

// 先学会写缓冲区
#define BUFFERSIZEDEFAULT 1024
#define INF 0
#define DBG 1
#define ERR 2

#define LOG_LEVEL DBG

#define LOG(level, format, ...)                                                               \
    do                                                                                        \
    {                                                                                         \
        if (level >= LOG_LEVEL)                                                               \
        {                                                                                     \
            time_t ti = time(nullptr);                                                        \
            struct tm *t = localtime(&ti);                                                    \
            char r[32];                                                                       \
            strftime(r, 31, "%H:%M:%S", t);                                                   \
            fprintf(stdout, "[%d %s %s %d] " format "\n", std::this_thread::get_id(), r, __FILE__, __LINE__, ##__VA_ARGS__); \
        }                                                                                     \
    } while (0)

#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__);
#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__);
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__);

class Buffer
{
private:
    std::vector<char> _buffer;
    int _writer_idx;
    int _reader_idx;

public:
    Buffer()
        : _buffer(BUFFERSIZEDEFAULT),
          _writer_idx(0),
          _reader_idx(0)
    {
    }

    void *Begin()
    {
        return &(*_buffer.begin());
    }

    void *WritePosition()
    {
        return (char *)Begin() + _writer_idx;
    }

    void *ReadPosition()
    {
        return (char *)Begin() + _reader_idx;
    }

    int WriteAbleSize()
    {
        return _buffer.size() - _writer_idx;
    }

    int ReadAbleSize()
    {
        return _writer_idx - _reader_idx;
    }

    int PrevReadSize()
    {
        return _reader_idx;
    }

    void WriterMove(int x)
    {
        _writer_idx += x;
    }

    void ReaderMove(int x)
    {
        _reader_idx += x;
    }

    void Write(void *data, size_t len)
    {
        EnsureWriteAble(len);
        std::copy((char *)data, (char *)data + len, (char *)WritePosition());
    }

    void WriteAndMove(void *data, size_t len)
    {
        Write(data, len);
        WriterMove(len);
    }

    void WriteString(std::string str)
    {
        Write((void *)str.c_str(), str.size());
    }

    void WriteStringAndMove(std::string str)
    {
        WriteString(str);
        WriterMove(str.size());
    }

    void WriteBuffer(Buffer buf)
    {
        Write(buf.ReadPosition(), buf.ReadAbleSize());
    }

    void WriteBufferAndMove(Buffer buf)
    {
        WriteBuffer(buf);
        WriterMove(buf.ReadAbleSize());
    }

    void Read(void *data, size_t len)
    {
        std::copy((char *)ReadPosition(), (char *)ReadPosition() + len, (char *)data);
    }

    void ReadAndMove(void *data, size_t len)
    {
        Read(data, len);
        ReaderMove(len);
    }

    std::string ReadString(size_t len)
    {
        assert(len <= ReadAbleSize());
        std::string ret;
        ret.resize(len);
        Read((void *)&(ret[0]), ret.size());
        return ret;
    }

    std::string ReadStringAndMove(size_t len)
    {
        std::string ret = ReadString(len);
        ReaderMove(len);
        return ret;
    }

    char *FindCRLF()
    {
        void *ret = memchr(ReadPosition(), '\n', ReadAbleSize());
        return (char*)ret;
    }

    std::string GetLine()
    {
        char *end = FindCRLF();
        if (end == nullptr)       //没有找到'\n'直接就返回空字符串了， 不会找到文件末尾
        {
            return "";
        }

        return ReadString(end - (char*)ReadPosition() + 1);
    }

    std::string GetLineAndMove()
    {
        std::string ret = GetLine();
        ReaderMove(ret.size());
        return ret;
    }

    void clear()
    {
        _writer_idx = 0;
        _reader_idx = 0;
    }

private:
    void EnsureWriteAble(int len)
    {
        if (len < WriteAbleSize())
            return;                                  // 不需要扩容
        if (PrevReadSize() + WriteAbleSize() >= len) // 将缓冲区拷贝到开始位置后空间足够
        {
            std::copy((char *)ReadPosition(), (char *)WritePosition(), (char *)Begin()); // 将有效数据都拷贝到一开始
            return;
        }
        _buffer.resize(((char *)WritePosition() - (char *)Begin()) + len);
    }
};

class Socket
{
private:
    int _sockfd;

public:
    Socket() {}

    Socket(int sockfd)
        : _sockfd(sockfd)
    {
    }
    //

    int Fd()
    {
        return _sockfd;
    }

    void Create()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (_sockfd < 0)
        {
            ERR_LOG("create socket error");
            abort();
        }
    }

    void Bind(uint16_t port, std::string ip)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        if (bind(_sockfd, (const sockaddr *)&addr, sizeof(addr)) < 0)
        {
            ERR_LOG("bind error");
            abort();
        }
    }

    bool Listen()
    {
        int ret = listen(_sockfd, 10);
        if (ret < 0)
        {
            ERR_LOG("listen error");
            return false;
        }
        return true;
    }

    bool Connection(uint16_t port, std::string &ip)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        int ret = connect(_sockfd, (sockaddr *)&addr, len);
        if (ret < 0)
        {
            ERR_LOG("connect error");
            return false;
        }
        return true;
    }

    int Accept()
    {
        int ret = accept(_sockfd, NULL, NULL);
        if (ret < 0)
        {
            ERR_LOG("accept error");
            return -1;
        }
        return ret;
    }

    // 从缓冲区中读取数据
    int Read(void *data, uint64_t len, int flag = 0)
    {
        int n = recv(_sockfd, data, len, flag);
        if (n <= 0)
        {
            if (errno == EAGAIN || errno == EINTR) // 说明是信号中断或者非阻塞
            {
                return 0;
            }
            ERR_LOG("read error");
            return -1;
        }
        return n;
    }

    int NonBlockRead(void *data, uint64_t len)
    {
        return Read(data, len, MSG_DONTWAIT);
    }

    int Send(void *data, uint64_t len, int flag = 0)
    {
        int n = send(_sockfd, data, len, flag);
        if (n < 0)
        {
            ERR_LOG("send error");
            return -1;
        }
        return n;
    }

    int NonBlockSend(void *data, uint64_t len)
    {
        return Send(data, len, MSG_DONTWAIT);
    }

    void Close()
    {
        if (_sockfd != -1)
        {
            ERR_LOG("关闭fd成功");
            close(_sockfd);
            _sockfd = -1;
        }
    }

    // 创建一个新的服务端连接
    bool CreateServer(uint16_t port, std::string ip = "0.0.0.0", int block_flag = 0)
    {
        // 创建套接字， 是否设置非阻塞， 绑定地址信息， 监听新连接， 启动地址重用。
        Create();
        if (block_flag == 1)
            SetNonBLock();
        else
            SetBLock();
        Bind(port, ip);
        Listen();
        SetReuse();
        return true;
    }

    bool CreateClient(uint16_t port, std::string ip)
    {
        Create();
        Connection(port, ip);
        return true;
    }

    bool SetNonBLock()
    {
        int fl = fcntl(_sockfd, F_GETFL);
        if (fl < 0)
        {
            ERR_LOG("get fl error");
            abort();
        }
        fcntl(_sockfd, F_SETFL, fl | O_NONBLOCK);
        return true;
    }

    bool SetBLock()
    {
        int fl = fcntl(_sockfd, F_GETFL);
        if (fl < 0)
        {
            ERR_LOG("get fl error");
            abort();
        }
        fcntl(_sockfd, F_SETFL, fl & ~O_NONBLOCK);
        return true;
    }

    void SetReuse()
    {
        int val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&val, sizeof(int));
        val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void *)&val, sizeof(int));
    }
};

class Poller;
class EventLoop;

class Channel
{
private:
    using EventCallback = std::function<void()>;

private:
    int _fd;
    EventLoop *_loop;
    uint32_t _event;  // 要监控的事件
    uint32_t _revent; // 已经就绪的事件

    EventCallback _read_eventcallback;
    EventCallback _write_eventcallback;
    EventCallback _close_eventcallback;
    EventCallback _error_eventcallback;
    EventCallback _event_eventcallback;

public:
    Channel(int fd, EventLoop *loop)
        : _fd(fd),
          _loop(loop)
    {
    }

    void SetReadCallback(const EventCallback &cb)
    {
        _read_eventcallback = cb;
    }

    void SetWriteCallback(const EventCallback &cb)
    {
        _write_eventcallback = cb;
    }

    void SetCloseCallback(const EventCallback &cb)
    {
        _close_eventcallback = cb;
    }

    void SetErrorCallback(const EventCallback &cb)
    {
        _error_eventcallback = cb;
    }

    void SetEventCallback(const EventCallback &cb)
    {
        _event_eventcallback = cb;
    }

    int Fd()
    {
        return _fd;
    }

    bool ReadAble()
    {
        return _event & EPOLLIN;
    }

    bool WriteAble()
    {
        return _event & EPOLLOUT;
    }

    uint32_t GetEvent()
    {
        return _event;
    }

    void SetREvent(uint32_t newevent) // 设置实际就绪的事件
    {
        _revent = newevent;
    }

    void EnableRead()
    {
        _event |= EPOLLIN;
        Updata();
    }

    void EnableWrite()
    {
        _event |= EPOLLOUT;
        Updata();
    }

    void DisableRead()
    {
        _event &= ~EPOLLIN;
        Updata();
    }

    void DisableWrite()
    {
        _event &= ~EPOLLOUT;
        Updata();
    }

    void DisableAll()
    {
        _event = 0;
        Updata();
    }
    void Updata(); // 更新epoll模型
    void Remove(); // 移除epoll模型中标记

    void HandleEvent()
    {
        // DBG_LOG("HandleEvent()");
        if (_revent & EPOLLIN || _revent & EPOLLRDHUP || _revent & EPOLLPRI)
        {
            DBG_LOG("EPOLLIN");
            if (_read_eventcallback)
            {
                _read_eventcallback();
            }
        }
        else if (_revent & EPOLLOUT)
        {
            DBG_LOG("EPOLLOUT");

            if (_write_eventcallback)
            {
                _write_eventcallback();
            }
        }
        else if (_revent & EPOLLERR)
        {
            if (_error_eventcallback)
            {
                _error_eventcallback();
            }
        }
        else if (_revent & EPOLLHUP)
        {
            if (_close_eventcallback)
            {
                _close_eventcallback();
            }
        }

        if (_event_eventcallback)
            _event_eventcallback();
    }
};

#define MAXEVENTSIZEDEFAULT 1024

class Poller
{
private:
    int _epfd;
    struct epoll_event events[MAXEVENTSIZEDEFAULT];
    std::unordered_map<int, Channel *> _fd_chns;

private:
    void UpData(int op, struct epoll_event *ev)
    {
        int fd = ev->data.fd;
        int ret = epoll_ctl(_epfd, op, fd, ev);
        if (ret < 0)
        {
            DBG_LOG("update error: %d", fd);    


            ERR_LOG("epoll updata error");
        }
    }

    bool HasChannel(Channel *ch)
    {
        int fd = ch->Fd();

        auto it = _fd_chns.find(fd);

        if (it == _fd_chns.end())
        {
            return false;
        }
        return true;
    }

public:
    Poller()
    {
        _epfd = epoll_create(10000);
        if (_epfd < 0)
        {
            
            ERR_LOG("epoll create error");
        }
    }

    void UpdataEvent(Channel *chn)
    {
        DBG_LOG("UpdataEvent");
        // 如果存在就更新， 不存在就添加
        if (HasChannel(chn)) // 存在就修改
        {
            struct epoll_event ev;
            ev.data.fd = chn->Fd();
            ev.events = chn->GetEvent();

            UpData(EPOLL_CTL_MOD, &ev);
        }
        else
        {
            struct epoll_event ev;
            ev.data.fd = chn->Fd();
            ev.events = chn->GetEvent();

            UpData(EPOLL_CTL_ADD, &ev);
            _fd_chns[chn->Fd()] = chn;
        }
    }

    void RemoveEvent(Channel *chn)
    {
        if (HasChannel(chn) == false)
        {
            return;
        }
        //
        int fd = chn->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        UpData(EPOLL_CTL_DEL, &ev);
        close(fd);
    }

    void Poll(std::vector<Channel *> *active)
    {
        int ret = epoll_wait(_epfd, events, MAXEVENTSIZEDEFAULT, -1);
        if (ret < 0)
        {
            if (errno == EINTR)
            {
                ERR_LOG("sign interupt : %s", strerror(errno));
            }
            else
            {
                ERR_LOG("poll error");
            }
            abort();
        }
        else if (ret == 0)
        {
            DBG_LOG("none event ready");
        }
        else
        {
            for (int i = 0; i < ret; i++)
            {
                auto it = _fd_chns.find(events[i].data.fd);
                if (it != _fd_chns.end())
                {
                    it->second->SetREvent(events[i].events);
                    active->push_back(it->second); // 这里是将要处理的事件加入到就绪数组中交给上层处理。但是对于就绪事件的处理还没有修改
                }
                else
                {
                    ERR_LOG("epoll and _fd_chns unMatch");
                    abort();
                }
            }
            DBG_LOG("处理了事件个数为 : %d", ret);
        }
    }
};

// using TaskFunc = std::function<void()>;
// using ReleaseFunc = std::function<void()>;

// class TimeTask
// {
// private:
//     uint64_t _id;
//     uint32_t _timeout;
//     TaskFunc _task;
//     ReleaseFunc _releasefunc;

// public:
//     TimeTask(uint64_t id, uint32_t delay, const TaskFunc& task)
//         : _id(id), _timeout(delay), _task(task)
//     {
//     }
//     ~TimeTask()
//     {
//         _task();
//         _releasefunc();
//     }

//     void SetReleaseFunc(const ReleaseFunc &rf)
//     {
//         _releasefunc = rf;
//     }

//     uint32_t GetDelay()
//     {
//         return _timeout;
//     }
// };

// class TimeWheel
// {
// private:
//     int _tick;
//     int _capacity;
//     std::vector<std::vector<std::shared_ptr<TimeTask>>> _time_wheel;
//     std::unordered_map<uint64_t, std::weak_ptr<TimeTask>> _map_task;

//     void RemoveTask(uint64_t id)
//     {
//         _map_task.erase(id);
//     }

// public:
//     TimeWheel()
//         : _tick(0), _capacity(60), _time_wheel(60, std::vector<std::shared_ptr<TimeTask>>())
//     {
//     }

//     void TimeAdd(uint64_t id, uint32_t delay, const TaskFunc &timetask)
//     {
//         _time_wheel[_tick].push_back(std::make_shared<TimeTask>(id, delay, timetask));
//         _map_task[id] = _time_wheel[_tick].back();
//         _map_task[id].lock()->SetReleaseFunc(std::bind(&TimeWheel::RemoveTask, this, id));
//     }

//     void RefreshTime(uint64_t id)
//     {
//         auto it = _map_task.find(id);
//         if (it == _map_task.end())
//         {
//             ERR_LOG("can find task, error");
//             abort();
//         }
//         std::shared_ptr<TimeTask> st = it->second.lock();
//         int delay = st->GetDelay();
//         int pos = (_tick + delay) % _capacity;

//         _time_wheel[pos].push_back(st);
//     }

//     void OnStep()
//     {
//         _time_wheel[_tick].clear();
//         _tick += 1;
//     }

// };

/*对于时间轮， 要管理一个定时器任务管理对象。 对于这个定时器任务管理对象，
他要有自己的定期任务id，也要有自己的超时时间， 有自己的任务。*/
using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;

class TimeTask
{
private:
    uint64_t _id;
    uint32_t _timeout;
    TaskFunc _task;
    ReleaseFunc _rf;
    bool _cancel;

public:
    TimeTask(uint64_t id, uint32_t timeout, TaskFunc task)
        : _id(id), _timeout(timeout), _task(task),
          _cancel(false)
    {
    }

    ~TimeTask()
    {
        if (_cancel == false)
            _task();
        _rf();
    }

    void Setrf(const ReleaseFunc &rf) { _rf = rf; }
    uint32_t GetDelay()
    {
        return _timeout;
    }

    void Cancel()
    {
        _cancel = true;
    }
};

class TimeWheel
{
public:
    using SPtrTask = std::shared_ptr<TimeTask>;
    using WPtrTask = std::weak_ptr<TimeTask>;

private:
    int _tick;
    int _capacity;
    std::vector<std::vector<SPtrTask>> _tasks;
    std::unordered_map<uint64_t, WPtrTask> _map_task;

    EventLoop *_loop; // 每一个管理对象内部都要有个loop,可以理解为父指针， 确保自己对外提供的操作能够在所在的
                      // 线程内执行。
    int _timerfd;
    std::unique_ptr<Channel> _timerfd_channel;

    void ReleaseTask(uint64_t id)
    {
        _map_task.erase(id);
    }

    int TimerfdCreate()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
        if (timerfd < 0)
        {
            ERR_LOG("timerfd create error");
            abort();
        }

        struct itimerspec item;
        item.it_interval.tv_sec = 1;
        item.it_interval.tv_nsec = 0;
        item.it_value.tv_sec = 1;
        item.it_value.tv_nsec = 0;

        int tmp = timerfd_settime(timerfd, 0, &item, nullptr);
        if (tmp < 0)
        {
            ERR_LOG("timerfd setting error");
            abort();
        }

        return timerfd;
    }

    void Readtimerfd()
    {
        uint64_t timer;
        int ret = read(_timerfd, &timer, sizeof(timer));
        if (ret < 0)
        {
            ERR_LOG("timerfd read error");
            abort();
        }
    }

    void OneStep()
    {
        _tasks[_tick].clear();
        _tick += 1;
    }

    void OnTime()
    {
        Readtimerfd();
        OneStep();
    }

    void TimeAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &task)
    {
        SPtrTask sptr = std::make_shared<TimeTask>(id, delay, task); /*创建一个定时器对象*/
        sptr->Setrf(std::bind(&TimeWheel::ReleaseTask, this, id));   /*给定时器添加销毁任务*/
        _tasks[_tick].push_back(sptr);                               /*将定时器添加到时间轮中*/
        WPtrTask wptr(sptr);                                         /*使用弱指针保存一下定时器方便后续定时器的使用，不增加pcount*/
        _map_task[id] = wptr;                                        /*将定时器对象利用弱指针管理，并将弱指针交给_map管理*/
    }

    void TimeRefreshInLoop(uint64_t id)
    {
        auto it = _map_task.find(id); /*刷新定时器的活跃状态就是将定时器往时间轮的后delay个单位再保存一个。*/
        if (it == _map_task.end())
        {
            ERR_LOG("timerefresh find id error, you timetask not add");
            abort();
        }
        SPtrTask sptr = it->second.lock(); /*拿到id是为了找到定时器的对象， 然后能够利用定时器对象*/
        uint32_t delay = sptr->GetDelay(); /*获得定时器延迟后在时间轮中的位置*/
        int pos = (_tick + delay) % _capacity;
        _tasks[pos].push_back(sptr);
    }

    void CancelInLoop(uint64_t id)
    {
        auto it = _map_task.find(id);
        if (it != _map_task.end())
        {

            auto pt = it->second.lock();

            if (pt != nullptr)
            {
                pt->Cancel();
            }
        }
    }

public:
    TimeWheel(EventLoop *loop)
        : _tick(0), _capacity(60), _tasks(_capacity),
          _timerfd(TimerfdCreate()), _loop(loop),
          _timerfd_channel(new Channel(_timerfd, _loop))
    {
        _timerfd_channel->SetReadCallback(std::bind(&TimeWheel::OnTime, this));
        _timerfd_channel->EnableRead(); // 时间轮设置可读事件。
    }

    void TimeAdd(uint64_t id, uint32_t delay, const TaskFunc &task);

    void TimeRefresh(uint64_t id);

    void TimeCancel(uint64_t id);

    // 有线程安全问题
    bool HasTimer(uint64_t id)
    {
        auto it = _map_task.find(id);
        if (it != _map_task.end())
        {
            return true;
        }
        return false;
    }
};

class EventLoop
{
public:
    using TaskCallback = std::function<void()>;

private:
    Poller _poller;
    std::thread::id _thread_id;
    int _eventfd;
    std::unique_ptr<Channel> _event_channel;
    TimeWheel _time_wheel;
    std::vector<TaskCallback> _tasks;
    std::mutex _mutex;

private:
    int CreateEventfd()
    {
        int ret = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        return ret;
    }

    void EventReadCallback()
    {
        uint64_t i = 0;
        int ret = recv(_eventfd, &i, sizeof(i), 0);
        if (ret < 0)
        {
            ERR_LOG("recv eventfd error");
            abort();
        }
    }

    void EventWeakUpCallback()
    {
        uint64_t i = 0;
        int ret = send(_eventfd, &i, sizeof(i), 0);
        if (send < 0)
        {
            ERR_LOG("write to eventfd error");
            abort();
        }
    }

    bool IsInLoop()
    {
        return std::this_thread::get_id() == _thread_id;
    }

    void QueueInLoop(const TaskCallback &cb)
    {
        {
            std::unique_lock<std::mutex> lock(_mutex); // 任务池是临界资源， 需要加锁保护。

            _tasks.push_back(cb);
        }
        EventWeakUpCallback();
    }

    void RunAllTask()
    {
        std::vector<TaskCallback> taskcallbacks;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            taskcallbacks.swap(_tasks);
        }
        for (auto &e : taskcallbacks)
        {
            e();
        }
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()),
          _eventfd(CreateEventfd()),
          _event_channel(new Channel(_eventfd, this)),
          _time_wheel(this)
    {
        _event_channel->SetReadCallback(std::bind(&EventLoop::EventReadCallback, this));
        _event_channel->EnableRead();
    }

    void RunInLoop(const TaskCallback &cb)
    {
        // loop的RunInLoop操作
        if (IsInLoop())
        {
            cb();
            return;
        }

        QueueInLoop(cb);
        return;
    }

    void UpdataEvent(Channel *chn)
    {
        _poller.UpdataEvent(chn);
    }

    void RemoveEvent(Channel *chn)
    {
        _poller.RemoveEvent(chn);
    }

    void TimerAdd(uint64_t id, int delay, const TaskFunc &cb)
    {
        _time_wheel.TimeAdd(id, delay, cb);
    }

    void TimerCancel(uint64_t id)
    {
        _time_wheel.TimeCancel(id);
    }

    void TimeRefresh(uint64_t id)
    {
        _time_wheel.TimeRefresh(id);
    }

    bool HasTimer(uint64_t id)
    {
        return _time_wheel.HasTimer(id);
    }

    void Start()
    {
        while (1)
        {
            std::vector<Channel *> actives;
            _poller.Poll(&actives);

            for (auto &e : actives)
            {
                e->HandleEvent();
            }

            RunAllTask();
        }
    }
};

void Channel::Updata() // 更新epoll模型
{
    _loop->UpdataEvent(this);
}
void Channel::Remove() // 移除epoll模型中标记
{
    _loop->RemoveEvent(this);
}

void TimeWheel::TimeAdd(uint64_t id, uint32_t delay, const TaskFunc &task)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimeAddInLoop, this, id, delay, task));
}

void TimeWheel::TimeRefresh(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimeRefreshInLoop, this, id));
}

void TimeWheel::TimeCancel(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::CancelInLoop, this, id));
}

typedef enum
{
    DISCONNECTED, // 连接关闭状态
    CONNECTING,   // 连接中状态
    CONNECTED,    // 连接完毕状态
    DISCONNECTING // 连接关闭中状态
} Connstate;

class Connection;
using PtrConn = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
public:
    using _connect_callback = std::function<void(const PtrConn &)>;
    using _message_callback = std::function<void(const PtrConn &, Buffer *)>; // 确保对象不会被意外销毁。
    using _anyevent_callback = std::function<void(const PtrConn &)>;
    using _close_callback = std::function<void(const PtrConn &)>;

private:
    // 当前连接的id
    int _conn_id;
    
    /*EventLoop管理者*/
    EventLoop *_loop;
    /*套接字的管理*/
    int _sockfd;
    Socket _socket;

    /*事件监控的管理*/
    Channel _sockfd_channel;

    /*上下文的管理*/
    std::any _context;

    /*缓冲区的管理*/
    Buffer _in_buffer;
    Buffer _out_buffer;


    // 连接状态的管理
    Connstate _state;
    bool _enable_inactive_destroy;

    _message_callback _mesg_call;
    _connect_callback _conn_call;
    _anyevent_callback _any_call;
    _close_callback _close_call;
    _close_callback _serverclose_call;

private:
    void HandleRead()
    {
        char buffer[65535];
        int n = recv(_sockfd, buffer, sizeof(buffer), 0);
        if (n < 0)    //客户端崩溃， 网络连接断开， 服务器不要关
        {
            ERR_LOG("handle Read() : recv error");
            ShutDownInLoop();
        }
        if (n == 0)
        {
            _sockfd_channel.DisableRead();
            _sockfd_channel.DisableAll();
            ShutDownInLoop();
        }
        buffer[n] = 0;
        _in_buffer.WriteAndMove(buffer, n);
        if (_in_buffer.ReadAbleSize() > 0)
        {
            _mesg_call(shared_from_this(), &_in_buffer);
        }
    }

    void HandleWrite()
    {
        int ret = _socket.Send(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
        if (ret < 0)
        {
            ERR_LOG("handleWrite error");
            if (_in_buffer.ReadAbleSize() > 0)
                _mesg_call(shared_from_this(), &_in_buffer);
            Release();
        }
        else if (ret == 0)
        {
            _sockfd_channel.DisableWrite();
        }
        else
        {
            DBG_LOG("Write over, %d, %d", ret, _out_buffer.ReadAbleSize());
            _out_buffer.ReaderMove(ret);
            if (_out_buffer.ReadAbleSize() == 0)
                _sockfd_channel.DisableWrite();
        }

        if (_state == DISCONNECTING)
        {
            Release();
        }
    }

    void HandleError()
    {
        HandleClose();
    }

    void HandleClose()
    {
        if (_in_buffer.ReadAbleSize() > 0)
        {
            _mesg_call(shared_from_this(), &_in_buffer);
        }

        Release();
    }

    void HandleEvent()
    {
        if (_enable_inactive_destroy == true)
        {
            _loop->TimeRefresh(_conn_id);
        }
        if (_any_call)
        {
            _any_call(shared_from_this());
        }
    }

    void EstablistedInLoop()
    {
        if (_state == CONNECTING)
        {
            _state = CONNECTED;
            DBG_LOG("___________________________________2");
            _sockfd_channel.EnableRead();

            if (_conn_call)
                _conn_call(shared_from_this());
            if (_any_call)
                _any_call(shared_from_this());
            return;
        }
    }

    void SendInLoop(void *data, size_t len)
    {
        if (_state == DISCONNECTED)
            return;

        _out_buffer.WriteAndMove(data, len);
        DBG_LOG("%s", (char*)_out_buffer.ReadPosition());

        if (!_sockfd_channel.WriteAble())
        {
            _sockfd_channel.EnableWrite();
        }
    }

    void ReleaseInLoop()
    {
        DBG_LOG("release a connetion: %d", _conn_id);
        _state = DISCONNECTED;

        // 移除epoll监控、关闭文件描述符、关闭非活跃销毁, 设置关闭连接状态
        _sockfd_channel.Remove();

        // close(_sockfd);
        _socket.Close();
        if (_loop->HasTimer(_conn_id))
            DisableInactiveDestroy();

        if (_close_call)
            _close_call(shared_from_this());
        if (_serverclose_call)
            _serverclose_call(shared_from_this());
    }

    void Release()
    {
        _loop->RunInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }

    void ShutDownInLoop()
    {
        _state = DISCONNECTING;
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_mesg_call)
                _mesg_call(shared_from_this(), &_in_buffer);
        }

        if (_out_buffer.ReadAbleSize() > 0)
        {
            if (_sockfd_channel.WriteAble() == false)
            {
                _sockfd_channel.EnableWrite();
            }
        }

        if (_out_buffer.ReadAbleSize() == 0)
        {
            Release();
        }
    }

    void EnableInactiveDestroyInLoop(int sec)
    {
        if (_loop->HasTimer(_conn_id))
        {
            _loop->TimeRefresh(_conn_id);
        }
        else
        {
            _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
        }

        _enable_inactive_destroy = true;
    }

    void DisableInactiveDestroyInLoop()
    {
        _enable_inactive_destroy = false;
        _loop->TimerCancel(_conn_id);
    }

    void UpGradeInLoop(const std::any &context, _message_callback &mcb, _connect_callback &ecb,
                       _anyevent_callback &acb, _close_callback &ccb)
    {
        _context = context;
        _mesg_call = mcb;
        _conn_call = ecb;
        _any_call = acb;
        _close_call = ccb;
    }

public:
    Connection(EventLoop *loop, int conn_id, int sockfd)
        : _loop(loop), _conn_id(conn_id), _sockfd(sockfd),
          _sockfd_channel(_sockfd, _loop),
          _state(CONNECTING),
          _enable_inactive_destroy(false),
          _socket(_sockfd)
    {
        _sockfd_channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _sockfd_channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _sockfd_channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _sockfd_channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _sockfd_channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
    }

    int Fd()
    {
        return _sockfd;
    }

    int Id()
    {
        return _conn_id;
    }

    bool IsConnected()
    {
        return _state == CONNECTED;
    }

    Connstate GetState()
    {
        return _state;
    }

    void SetContext(std::any context)
    {
        _context = context;
    }

    std::any &GetContext()
    {
        return _context;
    }

    void Establisted()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablistedInLoop, this));
    }

    /*向缓冲区中写数据*/
    void Send(void *data, size_t len)
    {
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, data, len));
    }

    /*关闭连接*/
    void ShutDown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }

    /*设置上下文*/
    void UpGrade(const std::any &context, _message_callback &mcb, _connect_callback &ecb,
                 _anyevent_callback &acb, _close_callback &ccb)
    {
        _loop->RunInLoop(std::bind(&Connection::UpGradeInLoop, this, context, mcb, ecb, acb, ccb));
    }

    /*启动非活跃连接销毁功能*/
    void EnableInactiveDestroy(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveDestroyInLoop, this, sec));
    }

    /*取消非活跃连接销毁功能*/
    void DisableInactiveDestroy()
    {
        _loop->RunInLoop(std::bind(&Connection::DisableInactiveDestroyInLoop, this));
    }

    void SetMessageCallBack(const _message_callback &cb)
    {
        _mesg_call = cb;
    }

    void SetConnectedCallBack(const _connect_callback &cb)
    {
        _conn_call = cb;
    }

    void SetAnyEventCallBack(const _anyevent_callback &cb)
    {
        _any_call = cb;
    }

    void SetCloseCallBack(const _close_callback &cb)
    {
        _close_call = cb;
    }

    void SetServerCloseCallBack(const _close_callback &cb)
    {
        _serverclose_call = cb;
    }
};

class Acceptor
{
private:
    using AcceptorCallback = std::function<void(int)>;
    EventLoop *_loop;

    int _sockfd;

    Channel _accept_channel;

    Socket _socket;

    AcceptorCallback _accept_callback;

    void HandleRead()
    {
        int newfd = accept(_sockfd, nullptr, nullptr);
        if (newfd < 0)
        {
            ERR_LOG("Acceptor error");
            return;
        }
        DBG_LOG("acceptor success, fd : %d", newfd);
        if (_accept_callback)
            _accept_callback(newfd);
    }

    int createServer(uint16_t port)
    {
        int ret = _socket.CreateServer(port);
        if (ret < 0)
        {
            ERR_LOG("create server error");
            abort();
        }
        return _socket.Fd();
    }

public:
    Acceptor(EventLoop *loop, uint16_t port)
        : _loop(loop), _sockfd(createServer(port)),
          _socket(_sockfd), _accept_channel(_sockfd, _loop)
    {
        _accept_channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }

    void SetAcceptCallback(const AcceptorCallback &cb)
    {
        _accept_callback = cb;
    }

    void Listen()
    {
        _accept_channel.EnableRead();
    }
};

class LoopThread
{
private:
    std::thread _thread;
    EventLoop *_loop;
    std::mutex _mutex;
    std::condition_variable _condition;

    void ThreadEntry()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);

            _loop = &loop;

            _condition.notify_all();
        }

        _loop->Start();
    }

public:
    LoopThread()
        : _thread(&LoopThread::ThreadEntry, this),
          _loop(nullptr)
    {
    }

    EventLoop *GetLoop()
    {
        // 这里的意义是什么。 是为了防止返回_loop时， _loop还没有被子线程初始化。
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _condition.wait(lock, [&]()
                            { return _loop != nullptr; });
        }

        return _loop;
    }
};

class LoopThreadPool
{
private:
    int _thread_count;
    int _next_index;
    EventLoop *_base_loop;
    std::vector<LoopThread *> _threads;
    std::vector<EventLoop *> _loops;

public:
    LoopThreadPool(EventLoop *base_loop)
        : _base_loop(base_loop),
          _next_index(0),
          _thread_count(0)
    {
    }

    void SetThreadCount(int count)
    {
        _thread_count = count;
    }

    void CreateThread()
    {
        _threads.resize(_thread_count);
        _loops.resize(_thread_count);

        for (int i = 0; i < _thread_count; i++)
        {
            _threads[i] = new LoopThread();
            _loops[i] = _threads[i]->GetLoop();
        }
    }

    EventLoop *_next_loop()
    {
        _next_index = (_next_index + 1) % _thread_count;
        return _loops[_next_index];
    }
};

class TcpServer
{
public:
    using _connect_callback = std::function<void(const PtrConn &)>;
    using _message_callback = std::function<void(const PtrConn &, Buffer *)>; // 确保对象不会被意外销毁。
    using _anyevent_callback = std::function<void(const PtrConn &)>;
    using _close_callback = std::function<void(const PtrConn &)>;

private:
    int _port;
    EventLoop _base_loop;

    bool _enable_inactive_destroy;
    int _timeout;

    uint64_t _conn_id;
    Acceptor _acceptor;

    LoopThreadPool _pool;
    std::unordered_map<uint64_t, PtrConn> _conns;

    _connect_callback _conn_call;
    _message_callback _mess_call;
    _anyevent_callback _any_call;
    _close_callback _close_call;
    _close_callback _serverclose_call;

    void RemoveConnectionCallback(PtrConn conn)
    {
        _base_loop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }

    void RemoveConnectionInLoop(PtrConn conn)
    {
        uint64_t id = conn->Id();
        auto it = _conns.find(id);
        if (it == _conns.end())
        {
            return;
        }
        _conns.erase(it);
    }

    void NewConnection(int _sockfd)
    {
        PtrConn newconn(new Connection(_pool._next_loop(), ++_conn_id, _sockfd));
        newconn->SetMessageCallBack(_mess_call);
        newconn->SetConnectedCallBack(_conn_call);
        newconn->SetCloseCallBack(_close_call);
        newconn->SetAnyEventCallBack(_any_call);
        newconn->SetServerCloseCallBack(std::bind(&TcpServer::RemoveConnectionCallback, this, newconn));
        if (_enable_inactive_destroy == true)
            newconn->EnableInactiveDestroy(_timeout);
        DBG_LOG("______________________________");
        newconn->Establisted();
        _conns[_conn_id] = newconn;
    }

    void AfterTaskInLoop(const TaskFunc &task, int delay)
    {
        _base_loop.TimerAdd(++_conn_id, delay, task);
    }

public:
    TcpServer(uint16_t port)
        : _enable_inactive_destroy(false),
          _port(port),
          _conn_id(0),
          _pool(&_base_loop),
          _timeout(0),
          _acceptor(&_base_loop, port)
    {
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.Listen();
    }

    void SetThreadPool(int count)
    {
        _pool.SetThreadCount(count);
        _pool.CreateThread();
    }

    void SetMessageCallBack(const _message_callback &cb)
    {
        _mess_call = cb;
    }

    void SetConnectedCallBack(const _connect_callback &cb)
    {
        _conn_call = cb;
    }

    void SetAnyEventCallBack(const _anyevent_callback &cb)
    {
        _any_call = cb;
    }

    void SetCloseCallBack(const _close_callback &cb)
    {
        _close_call = cb;
    }

    void EnableInactiveConnectDestroy(int delay)
    {
        _enable_inactive_destroy = true;
        _timeout = delay;
    }

    void AfterTask(const TaskFunc &task, int delay)
    {
        _base_loop.RunInLoop(std::bind(&TcpServer::AfterTaskInLoop, this, task, delay));
    }

    void Start()
    {
        _base_loop.Start();
    }
};